
package com.springboot.rmq.producer;

import com.google.common.primitives.Ints;
import java.util.Properties;

import com.springboot.rmq.comm.Constant;
import com.springboot.rmq.producer.bean.RmqMessage;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.util.StringUtils;


public class MsgProducer {

    /**
     * 日志记录
     */
    private static final Log logger = LogFactory.getLog(MsgProducer.class);

    /**
     * 实际使用的producer对象
     */
    private DefaultMQProducer producer;

    /**
     *
     */
    private Properties prop;

    public MsgProducer(Properties properties) {
        this.prop = properties;
        buildProducer();
    }

    /**
     * 创建producer并设置相关属性
     */
    private void buildProducer() {
        logger.info("start to create producer.");
        //根据每个业务模块，加载producer需要的配置文件


        //检查配置文件是否正确
        if (!validateProducerProp()) {
            logger.error("Producer Configuration is not right,please check Config.");
            return;
        }
        //配置文件中读取groupname
        String groupName = prop.getProperty(Constant.KEY_PRODUCER_GROUPNAME);
        //配置文件中读取nameserveradd
        String nameServerAddr = prop.getProperty(Constant.KEY_PRODUCER_NAMESERVER);
        //配置文件中读取topicqueue数量
        String topinQueueNum = prop.getProperty(Constant.KEY_PRODUCER_TOPICQUEUENUMS);
        //配置VIP频道发送消息
        String sendMsgWithVIPChannel = prop.getProperty(Constant.KEY_SEND_MSG_WITH_VIP_CHANNEL, "true");

        producer = new DefaultMQProducer(groupName);
        producer.setNamesrvAddr(nameServerAddr);
        producer.setProducerGroup(groupName);
        producer.setVipChannelEnabled(Boolean.parseBoolean(sendMsgWithVIPChannel));
        try {
            producer.setDefaultTopicQueueNums(Integer.parseInt(topinQueueNum));
        } catch (NumberFormatException e) {
            producer.setDefaultTopicQueueNums(4);
        }
        try {
            producer.setSendMsgTimeout(Integer.parseInt(prop.getProperty(Constant.KEY_PRODUCER_TIMEOUT)));
        } catch (NumberFormatException e) {
            producer.setSendMsgTimeout(3000);
        }
        //异步重试次数, 默认3次
        Integer asyncTime = Ints.tryParse(prop.getProperty(Constant.KEY_PRODUCER_RETRYASYNC_TIMES));
        producer.setRetryTimesWhenSendAsyncFailed(asyncTime == null ? 3 : asyncTime);
        //同步重试次数，默认3次
        Integer syncTime = Ints.tryParse(prop.getProperty(Constant.KEY_PRODUCER_RETRYSYNC_TIMES));
        producer.setRetryTimesWhenSendFailed(syncTime == null ? 3 : syncTime);

        //从配置文件中读取instance信息
        String instanceName = prop.getProperty(Constant.KEY_PRODUCER_INSTANCENAME);
        producer.setInstanceName(instanceName);

        logger.info("create producer over and ready to start it.");
        try {
            producer.start();
            logger.info("start producer successful with"
                + " instanceName[" + producer.getInstanceName() + "]");
        } catch (MQClientException e) {
            logger.error("start producer error.", e);
            return;

        }

    }

    /**
     * 检查producer参数是否正确
     *
     * @return true or false
     */
    private boolean validateProducerProp() {
        //检查nameserver
        String nameserver = prop.getProperty(Constant.KEY_PRODUCER_NAMESERVER);
        //必须是ip地址和端口
        if (!nameserver.matches(Constant.TOPIC_PATTEN)) {

            return false;
        }
        //检查producer group
        String groupname = prop.getProperty(Constant.KEY_PRODUCER_GROUPNAME);
        if (StringUtils.isEmpty(groupname)) {
            return false;
        }
        //检查主题队列数量
        String topicQueueNums = prop.getProperty(Constant.KEY_PRODUCER_TOPICQUEUENUMS);
        if (!StringUtils.isEmpty(topicQueueNums) && !topicQueueNums.matches("[0-9]+")) {
            return false;
        }
        //检查超时时间
        String timeout = prop.getProperty(Constant.KEY_PRODUCER_TIMEOUT);
        if (!StringUtils.isEmpty(timeout) && !timeout.matches("[0-9]+")) {
            return false;
        }
        return true;
    }


    /**
     * 同步方式，方式消息
     *
     * @param msg 需要发送的消息
     *
     * @return sendResult 同步发送结果
     * @throws Exception 异常由调用者进行处理判断
     */
    public SendResult sendSyncMsg(RmqMessage msg) throws Exception {
        Message sendMsg = new Message(msg.getTopic() /* Topic */,
            msg.getTags() /* Tag */, msg.getKey(),
            msg.getBody().getBytes(RemotingHelper.DEFAULT_CHARSET));/* Message body */
        SendResult sendResult = producer.send(sendMsg);
        return sendResult;
    }

    /**
     * 同步方式，方式消息，带重试3次
     *
     * @param msg 需要发送的消息
     *
     * @return sendResult 同步发送结果
     * @throws Exception 异常由调用者进行处理判断
     */
    public SendResult sendSyncMsgWithRetry(RmqMessage msg) throws Exception {
        return sendSyncMsgWithRetry(msg, 3);
    }

    /**
     * 同步方式，方式消息，带重试3次
     *
     * @param msg 需要发送的消息
     *
     * @return sendResult 同步发送结果
     * @throws Exception 异常由调用者进行处理判断
     */
    public SendResult sendSyncMsgWithRetry(RmqMessage msg, int retryTimes) throws Exception {
        Message sendMsg = new Message(msg.getTopic() /* Topic */,
            msg.getTags() /* Tag */, msg.getKey(),
            msg.getBody().getBytes(RemotingHelper.DEFAULT_CHARSET));/* Message body */
        int sendTime = 0;
        SendResult sendResult = null;
        while (sendTime < retryTimes) {
            try {
                sendResult = producer.send(sendMsg);
                logger.info("msg send result:" + sendResult);
                //发送成功
                break;
            } catch (Exception e) {
                logger.error("send message failed:", e);
                sendTime++;
                if (sendTime == retryTimes) {
                    logger.error("retry" + retryTimes + " still send failed");
                    throw e;
                }
            }
        }
        return sendResult;
    }

    /**
     * 同步顺序方式，方式消息
     *
     * @param msg      需要发送的消息
     * @param selector 顺序选择器，业务自己维护、
     * @param args1    顺序key
     *
     * @return sendResult 同步发送结果
     * @throws Exception 异常由调用者进行处理判断
     */
    public SendResult sendSyncMsgOrder(RmqMessage msg, MessageQueueSelector selector, Object args1) throws Exception {
        Message sendMsg = new Message(msg.getTopic() /* Topic */,
            msg.getTags() /* Tag */, msg.getKey(),
            msg.getBody().getBytes(RemotingHelper.DEFAULT_CHARSET));/* Message body */
        SendResult sendResult = producer.send(sendMsg, selector, args1);
        return sendResult;
    }

    /**
     * 同步顺序方式，方式消息，重试
     *
     * @param msg      需要发送的消息
     * @param selector 顺序选择器，业务自己维护、
     * @param args1    顺序key
     *
     * @return sendResult 同步发送结果
     * @throws Exception 异常由调用者进行处理判断
     */
    public SendResult sendSyncMsgOrderWithRetry(RmqMessage msg, MessageQueueSelector selector, Object args1) throws Exception {
        return sendSyncMsgOrderWithRetry(msg, selector, args1, 3);
    }

    /**
     * 同步顺序方式，方式消息,重试
     *
     * @param msg        需要发送的消息
     * @param selector   顺序选择器，业务自己维护、
     * @param args1      顺序key
     * @param retryTimes 重试次数
     *
     * @return sendResult 同步发送结果
     * @throws Exception 异常由调用者进行处理判断
     */
    public SendResult sendSyncMsgOrderWithRetry(RmqMessage msg, MessageQueueSelector selector, Object args1,
        int retryTimes) throws Exception {
        Message sendMsg = new Message(msg.getTopic() /* Topic */,
            msg.getTags() /* Tag */, msg.getKey(),
            msg.getBody().getBytes(RemotingHelper.DEFAULT_CHARSET));/* Message body */
        int sendTime = 0;
        SendResult sendResult = null;
        while (sendTime < retryTimes) {
            try {
                sendResult = producer.send(sendMsg, selector, args1);
                logger.info("msg send result:" + sendResult);
                //发送成功
                break;
            } catch (Exception e) {
                logger.error("send message failed:", e);
                sendTime++;
                if (sendTime == retryTimes) {
                    logger.error("retry" + retryTimes + " still send failed");
                    throw e;
                }
            }
        }
        return sendResult;

    }

    /**
     * 通过异步方式发送消息
     *
     * @param msg          需要发送的消息
     * @param sendCallback 回调函数
     *
     * @throws Exception 异常由调用者进行处理判断
     */
    public void sendAsyncMsg(RmqMessage msg, SendCallback sendCallback) throws Exception {
        Message sendMsg = new Message(msg.getTopic() /* Topic */,
            msg.getTags() /* Tag */, msg.getKey(),
            msg.getBody().getBytes(RemotingHelper.DEFAULT_CHARSET));/* Message body */
        producer.send(sendMsg, sendCallback);
    }

    /**
     * 通过异步顺序方式发送消息
     *
     * @param msg          需要发送的消息
     * @param selector     顺序选择器，业务自己维护、
     * @param args1        顺序key
     * @param sendCallback 回调函数
     *
     * @throws Exception 异常由调用者进行处理判断
     */
    public void sendAsyncMsgOrder(RmqMessage msg, MessageQueueSelector selector,
        Object args1, SendCallback sendCallback) throws Exception {
        Message sendMsg = new Message(msg.getTopic() /* Topic */,
            msg.getTags() /* Tag */, msg.getKey(),
            msg.getBody().getBytes(RemotingHelper.DEFAULT_CHARSET));/* Message body */
        producer.send(sendMsg, selector, args1, sendCallback);


    }

    /**
     * 关闭producer
     */
    public void closeProducer() {
        this.producer.shutdown();
    }
}
